package com.processor.base;

import com.ProcessorThreadFactory;
import com.handler.game.BaseHandler;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.event.ProcessorEvent;
import com.handler.base.DisruptorProcessorHandler;


/**
 * @author : gameZhao
 * @version V1.0
 * @Project: disruptorDemo
 * @Package handler
 * @Description: 利用disruptor 高效无锁特性,处理线程循环任务 多生产者,单消费者
 * @date Date : 2021年12月27日 11:07
 */
public abstract class DisruptorProcessor extends BaseProcessor {
    
    public static final int DEFAULT_SIZE = 8 * 1024;
    
    protected Disruptor<ProcessorEvent> disruptor;
    
    //使用EventTranslatorOneArg封装数据填充,这里使用lambda表达式
    private EventTranslatorOneArg<ProcessorEvent, BaseHandler> processorEventTranslator = (event, sequence, handler) -> {
        event.setHandler(handler);
    };
    
    /**
     * @param processorName
     */
    public DisruptorProcessor(String processorName, int ringBufferSize) {
        
        Disruptor disruptor = new Disruptor(ProcessorEvent::new, ringBufferSize,
                ProcessorThreadFactory.create(processorName), ProducerType.MULTI, new BlockingWaitStrategy());
        disruptor.handleEventsWith(new DisruptorProcessorHandler(processorName));
        disruptor.start();
    }
    
    /**
     * 单消费者构造器
     *
     * @param processorName
     */
    public DisruptorProcessor(String processorName) {
        this(processorName, DEFAULT_SIZE);
    }
    
    /**
     * 向disruptor提交事件
     *
     * @param handler
     */
    @Override
    public void executeHandler0(BaseHandler handler) {
        var ringBuffer = disruptor.getRingBuffer();
        handler.setTime(System.currentTimeMillis());
        ringBuffer.publishEvent(processorEventTranslator, handler);
    }
}
